package com.ds.infrastructure.hyperspace.container.endpoint.connector;

import com.ds.infrastructure.hyperspace.common.*;
import com.ds.infrastructure.hyperspace.common.config.TargetEndpointConfig;
import com.ds.infrastructure.hyperspace.common.config.TargetNodeConfig;
import com.ds.infrastructure.hyperspace.common.flow.DuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.endpoint.HyperspaceSuspendDataWrapper;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.TargetEndpointNode;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.TargetProtocolAdapter;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.flow.operation.EntryFlowOperation;
import com.ds.infrastructure.hyperspace.container.internal.HyperspaceIdGenerator;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import static com.ds.infrastructure.hyperspace.common.StateEnum.CLIENT_CLOSED_CONNECTION;
import static com.ds.infrastructure.hyperspace.common.StateEnum.TARGET_ENDPOINT_UNAVAILABLE;
import static com.ds.infrastructure.hyperspace.common.TargetEndpointStreamStatus.*;
import static com.ds.infrastructure.hyperspace.container.endpoint.NettyHandlerName.TARGET_HANDLER_ADAPTER_NAME;

/**
 * Note: this class is not allowed switch thread, you must run it on its own thread.
 *
 * @author  Weishaoying
 */
@Slf4j
public class NettyEndpointNodeStream implements TargetEndpointStream, Timeoutable {

    private final String streamId;

    private final NettyContainer nettyContainer;

    @Getter
    private final TargetEndpointNode targetEndpointNode;

    private final Bootstrap bootstrap;

    private final NettyEndpointNodeStreamManager nodeStreamManager;

    private EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation;

    private final HyperspaceConnector hyperspaceConnector;

    private final StreamConnectFutureListener streamConnectFutureListener;

    @Setter
    private TargetProtocolAdapter targetProtocolAdapter;

    @Getter
    private TargetEndpointStreamStatus status = TargetEndpointStreamStatus.INIT;

    private long lastActiveTime;

    private boolean preAcquire;

    private List<HyperspaceSuspendDataWrapper> suspendRequestList = new ArrayList<>();

    @Getter
    private InetSocketAddress remoteAddress;

    @Getter
    private boolean isOpenEntryEndpoint = true;

    @Getter
    @Setter
    private TargetEndpointStreamType streamType;

    public NettyEndpointNodeStream(boolean preAcquire,
                                   NettyEndpointNodeStreamManager nodeStreamManager) {
        this.streamId = HyperspaceIdGenerator.generate();
        this.preAcquire = preAcquire;
        this.nodeStreamManager = nodeStreamManager;
        this.hyperspaceConnector = nodeStreamManager.getHyperspaceConnector();
        this.entryFlowOperation = hyperspaceConnector.getEntryFlowOperation();
        this.nettyContainer = nodeStreamManager.getNettyContainer();
        this.bootstrap = nodeStreamManager.getBootstrap();
        this.targetEndpointNode = nodeStreamManager.getTargetEndpointNode();
        this.streamConnectFutureListener = new StreamConnectFutureListener(this);
        connect();
    }

    @Override
    public void connect() {
        if (isChannelActive() || CONNECTING == status) {
            nettyContainer.decrementTargetConnectionSize(targetEndpointNode);
            return;
        }
        status = CONNECTING;
        remoteAddress = targetEndpointNode.getResolverProvider().doSelect();
        if ((remoteAddress == null)) {
            onConnectFailed("no remote address!");
            return;
        }
        bootstrap.connect(remoteAddress).addListener(streamConnectFutureListener);
    }

    @Override
    public <C extends DuplexFlowContext> void handleInput(C duplexFlowContext, Object data) {
        if (status != ACQUIRED) {
            suspendRequestList.add(new HyperspaceSuspendDataWrapper(
                    (HyperspaceDuplexFlowContext) duplexFlowContext, data, false));
            return;
        }
        preHandleSuspendData();
        targetProtocolAdapter.writeToTarget((HyperspaceDuplexFlowContext) duplexFlowContext, data);
    }

    @Override
    public <C extends DuplexFlowContext> void handleInputComplete(C duplexFlowContext) {
        if (status != ACQUIRED) {
            suspendRequestList.add(new HyperspaceSuspendDataWrapper(
                    (HyperspaceDuplexFlowContext) duplexFlowContext, null, true));
            return;
        }
        preHandleSuspendData();
        targetProtocolAdapter.flushToTarget((HyperspaceDuplexFlowContext) duplexFlowContext);
    }

    @Override
    public void onTargetChannelClose() {
        if (CLOSED == status) {
            return;
        }
        status = CLOSED;
        nodeStreamManager.onStreamClose(this);
    }

    @Override
    public void changeTargetChannelReadable(boolean readable) {
        targetProtocolAdapter.changeTargetChannelReadable(readable);
    }

    @Override
    public void closeTargetChannel() {
        if (targetProtocolAdapter != null) {
            targetProtocolAdapter.closeTargetChannel();
            targetProtocolAdapter.clearSuspendData();
        }
        clearSuspendRequest();
    }

    @Override
    public <C extends DuplexFlowContext> void switchProtocol(
            C duplexFlowContext, EndpointProtocolEnum protocol) {
        targetProtocolAdapter.switchProtocol((HyperspaceDuplexFlowContext) duplexFlowContext, protocol);
    }

    @Override
    public void touch() {
        lastActiveTime = System.currentTimeMillis();
    }

    @Override
    public String getEventId() {
        return streamId;
    }

    @Override
    public boolean checkTimeout() {
        if (!isOpenEntryEndpoint) {
            return true;
        }
        if (!suspendRequestList.isEmpty()) {
            Iterator<HyperspaceSuspendDataWrapper> suspendDataWrapperIterator = suspendRequestList.iterator();
            while (suspendDataWrapperIterator.hasNext()) {
                HyperspaceDuplexFlowContext tempDuplexFlowContext = null;
                HyperspaceSuspendDataWrapper suspendDataWrapper = null;
                try {
                    suspendDataWrapper = suspendDataWrapperIterator.next();
                    tempDuplexFlowContext = suspendDataWrapper.getDuplexFlowContext();
                    if (isClosedDuplexFlowContext(tempDuplexFlowContext)) {
                        continue;
                    }
                    if (tempDuplexFlowContext.checkTimeout()) {
                        tempDuplexFlowContext.onTimeout();
                    }
                } finally {
                    if (tempDuplexFlowContext != null && isClosedDuplexFlowContext(tempDuplexFlowContext)) {
                        releaseDuplexFlowContextData(suspendDataWrapper);
                        suspendDataWrapperIterator.remove();
                    }
                }
            }
        }
        targetProtocolAdapter.checkTimeoutDuplexFlowContext();
        if (lastActiveTime < 1) {
            lastActiveTime = System.currentTimeMillis();
            return false;
        }
        return System.currentTimeMillis() - lastActiveTime >
                targetEndpointNode.getNodeConfig().getProtocolConfig().getConnectionMaxIdleTime();
    }

    @Override
    public void onTimeout() {
        closeTargetChannel();
    }

    @Override
    public boolean isAvailableTargetNode() {
        return targetEndpointNode.getResolverProvider().isAvailable(remoteAddress);
    }

    public void onConnectSuccess(TargetProtocolAdapter targetProtocolAdapter) {
        touch();
        this.targetProtocolAdapter = targetProtocolAdapter;
        preHandleSuspendData();
        if (this.preAcquire) {
            this.status = ACQUIRED;
            this.preAcquire = false;
        } else {
            this.status = IDLE;
        }
        this.nodeStreamManager.onStreamConnectSuccess(this);
    }

    public void releaseStreamWriteOperation() {
        if (!isChannelActive() || status != ACQUIRED) {
            return;
        }
        status = IDLE;
        this.nodeStreamManager.onStreamRelease(this);
    }

    boolean acquireWriteOperation() {
        if (IDLE == status && targetProtocolAdapter.isChannelActive()) {
            status = ACQUIRED;
            return true;
        }
        return false;
    }

    boolean isChannelActive() {
        return targetProtocolAdapter != null && targetProtocolAdapter.isChannelActive();
    }

    private void preHandleSuspendData() {
        if (suspendRequestList.isEmpty()) {
            return;
        }
        Iterator<HyperspaceSuspendDataWrapper> suspendDataWrapperIterator = suspendRequestList.iterator();
        while (suspendDataWrapperIterator.hasNext()) {
            try {
                HyperspaceSuspendDataWrapper suspendDataWrapper = suspendDataWrapperIterator.next();
                HyperspaceDuplexFlowContext tempDuplexFlowContext = suspendDataWrapper.getDuplexFlowContext();
                if (isClosedDuplexFlowContext(tempDuplexFlowContext)) {
                    releaseDuplexFlowContextData(suspendDataWrapper);
                    continue;
                }
                if (suspendDataWrapper.isFlush()) {
                    targetProtocolAdapter.flushToTarget(tempDuplexFlowContext);
                } else {
                    targetProtocolAdapter.writeToTarget(tempDuplexFlowContext, suspendDataWrapper.getData());
                }
            } finally {
                suspendDataWrapperIterator.remove();
            }
        }
    }

    private void onConnectFailed(String cause) {
        nettyContainer.decrementTargetConnectionSize(targetEndpointNode);
        nodeStreamManager.onStreamConnectFailed(this);
        status = CONNECT_FAILED;
        if (suspendRequestList.isEmpty()) {
            return;
        }
        Iterator<HyperspaceSuspendDataWrapper> suspendDataWrapperIterator = suspendRequestList.iterator();
        while (suspendDataWrapperIterator.hasNext()) {
            try {
                HyperspaceSuspendDataWrapper suspendDataWrapper = suspendDataWrapperIterator.next();
                HyperspaceDuplexFlowContext tempDuplexFlowContext = suspendDataWrapper.getDuplexFlowContext();
                if (isClosedDuplexFlowContext(tempDuplexFlowContext)) {
                    releaseDuplexFlowContextData(suspendDataWrapper);
                    continue;
                }
                if (tempDuplexFlowContext.getEndpointProtocol().isLongLive()) {
                    if (suspendDataWrapper.getData() == tempDuplexFlowContext.getRequestStartState()) {
                        entryFlowOperation.interruptAndCloseContext(tempDuplexFlowContext, TARGET_ENDPOINT_UNAVAILABLE,
                                "Target endpoint stream connect failed!,cause by " + cause);
                    } else {
                        releaseDuplexFlowContextData(suspendDataWrapper);
                    }
                } else {
                    TargetEndpoint<TargetEndpointConfig> targetEndpoint =
                            tempDuplexFlowContext.getTargetEndpoint();
                    List<EndpointNode<TargetNodeConfig>> nodeList = targetEndpoint.getEndpointNodes(
                            tempDuplexFlowContext.getZoneId());
                    boolean nodeAvailable = false;
                    for (EndpointNode<TargetNodeConfig> tempTargetEndpointNode : nodeList) {
                        if (tempTargetEndpointNode.isAvailable()) {
                            nodeAvailable = true;
                            break;
                        }
                    }
                    if (!nodeAvailable) {
                        String message = "Target endpoint stream connect failed! target node unavailable, endpoint name:"
                                + targetEndpoint.getNamespace();
                        log.warn(message);
                        entryFlowOperation.interruptAndCloseContext(tempDuplexFlowContext, TARGET_ENDPOINT_UNAVAILABLE,
                                message);
                        continue;
                    }
                    hyperspaceConnector.onContextWriteFailed(suspendDataWrapper.getDuplexFlowContext());
                }
            } finally {
                suspendDataWrapperIterator.remove();
            }
        }

    }

    private boolean isClosedDuplexFlowContext(HyperspaceDuplexFlowContext duplexFlowContext) {
        return duplexFlowContext.isClosed() || !duplexFlowContext.getEntryProtocolAdapter().isChannelOpen();
    }

    /**
     * 清除context中没有消化的数据
     *
     * @param suspendDataWrapper
     */
    private void releaseDuplexFlowContextData(HyperspaceSuspendDataWrapper suspendDataWrapper) {
        HyperspaceDuplexFlowContext duplexFlowContext = suspendDataWrapper.getDuplexFlowContext();
        Object data = suspendDataWrapper.getData();
        if (duplexFlowContext.isClosed() && data == duplexFlowContext.getRequestStartState()) {
            return;
        }
        duplexFlowContext.getEntryProtocolAdapter().releaseData(data);
    }

    @Override
    public void clearSuspendRequest() {
        if (!suspendRequestList.isEmpty()) {
            Iterator<HyperspaceSuspendDataWrapper> suspendDataWrapperIterator = suspendRequestList.iterator();
            while (suspendDataWrapperIterator.hasNext()) {
                HyperspaceSuspendDataWrapper suspendDataWrapper = suspendDataWrapperIterator.next();
                HyperspaceDuplexFlowContext duplexFlowContext = suspendDataWrapper.getDuplexFlowContext();
                if (!duplexFlowContext.isClosed() &&
                        suspendDataWrapper.getData() == duplexFlowContext.getRequestStartState()) {
                    entryFlowOperation.interruptAndCloseContext(duplexFlowContext, CLIENT_CLOSED_CONNECTION,
                            "Entry or target Client has closed connection! release data.");
                } else {
                    releaseDuplexFlowContextData(suspendDataWrapper);
                }
                suspendDataWrapperIterator.remove();
            }
        }
    }

    @Override
    public void noticeEntryChannelClosed() {
        isOpenEntryEndpoint = false;
    }

    private static class StreamConnectFutureListener implements GenericFutureListener<ChannelFuture> {

        private final NettyEndpointNodeStream stream;

        private static final int CONNECT_FAILED_LOG_STEP = 5;

        private AtomicInteger connectCount = new AtomicInteger();

        StreamConnectFutureListener(NettyEndpointNodeStream stream) {
            this.stream = stream;
        }

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!this.stream.suspendRequestList.isEmpty()) {
                this.stream.suspendRequestList.get(0).getDuplexFlowContext().endConnectionDuration();
            }
            Channel newChannel = future.channel();
            if (!future.isSuccess()) {
                if (connectCount.incrementAndGet() % CONNECT_FAILED_LOG_STEP == 0) {
                    TargetNodeConfig targetNodeConfig = stream.targetEndpointNode.getNodeConfig();
                    log.error("Connect to target host failed! target node {}:{}",
                            targetNodeConfig.getHost(), targetNodeConfig.getPort(), future.cause());
                    connectCount.set(0);
                }
                stream.onConnectFailed(String.valueOf(future.cause()));
                if (newChannel != null && newChannel.isOpen()) {
                    newChannel.flush().close();
                }
            } else {
                TargetProtocolAdapter targetProtocolAdapter = (TargetProtocolAdapter)
                        newChannel.pipeline().get(TARGET_HANDLER_ADAPTER_NAME);
                targetProtocolAdapter.attachStream(stream);
            }
        }
    }
}
